package com.prj.ufdm.demo.tutorial.hadoop.mr;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;

import com.prj.ufdm.demo.tutorial.hadoop.constants.HADOOP_HOST;
import com.prj.ufdm.demo.tutorial.hadoop.enums.MapReduceRunTypeEnum;
import com.prj.ufdm.demo.tutorial.hadoop.hdfs.HdfsClient;

/**
 * 通过Mapper Reducer 统计字频
 * @author Administrator
 */
public class WordCountJob {
	
	/**
	 * DESC: 运行 MapReduce
	 * 本地运行 mapreduce 需要本地设置环境变量(需要在WINDOWS设置HADOOP_HOME环境变量，Path 加入： %HADOOP_HOME%/bin)。
	 * 同时在 ~/bin 目录中。需要有 hadoop.dll，winutils.exe 两个文件，下载地址：https://github.com/cdarlint/winutils
	 * @auther 胡义振
	 * @date 2020年6月17日
	 * @param args
	 * @throws Exception 
	 * @return
	 */
	public static void main(String[] args) throws Exception {
		
		// 1.获取job任务
		Configuration conf = getConf();
		Job job = Job.getInstance(conf);
		// 2.获取jar包
		job.setJarByClass(WordCountJob.class);
		
		// 3.获取自定义的mapper与reducer类
		job.setMapperClass(WordCountMapper.class);
		job.setReducerClass(WordCountReducer.class);
		// 这里如果不设置，统计结果为0，官方实例中加这个
		job.setCombinerClass(WordCountReducer.class);
		
		// 4.设置map输出的数据类型
		job.setMapOutputKeyClass(Text.class);
		job.setMapOutputValueClass(IntWritable.class);
		
		// 5.设置reduce输出的数据类型（最终的数据类型）
		job.setOutputKeyClass(Text.class);
		job.setOutputValueClass(IntWritable.class);
		
		// 6.设置输入数据存在的路径与处理后的结果路径
		FileInputFormat.addInputPath(job, new Path(HADOOP_HOST.WORD_INPUT_PATH_FILE));
		FileOutputFormat.setOutputPath(job, new Path(HADOOP_HOST.WORD_OUTPUT_PATH));
		
		// 7. 提交任务
		HdfsClient.deleteOnExit(HADOOP_HOST.WORD_OUTPUT_PATH);
		
		// 7.1 远程自动提交到YARN上执行
		if(MapReduceRunTypeEnum.MAP_REDUCE_RUN_TYPE_1.getCode().equals(HADOOP_HOST.RUN_TYPE)) {
			// 这里的JAR包为本地JAR包路径，因为需要将jar包上传到hdfs上,如果没有设置就会找不到我们的业务实现类(ClassNotFondException)
			job.setJar("E:\\project_new\\UFDM\\ufdm-web-demo\\src\\main\\lib\\wc.jar");
			job.submit();
			System.out.println("\n 提交YARN执行 jobId:"+job.getJobID().toString());
		}
		// 7.2 本地运行（若提交到YARN上，需要把本地JAR包拷贝到服务器，通过命令执行- hadoop jar wc.jar com.prj.ufdm.demo.tutorial.hadoop.mr.WordCountJob ）
		else if(MapReduceRunTypeEnum.MAP_REDUCE_RUN_TYPE_0.getCode().equals(HADOOP_HOST.RUN_TYPE)){
			System.out.println("\n 本地执行..."); 
			boolean result = job.waitForCompletion(true);
			System.out.println("\n 本地执行 jobId:"+job.getJobID().toString() + ", 结果："+result);
			System.exit(result ? 0 : 1); 
		}
		
	}
	
	public static Configuration getConf() {
		
		Configuration conf = new Configuration();
		
		System.setProperty("HADOOP_USER_NAME","root");
		
		if(MapReduceRunTypeEnum.MAP_REDUCE_RUN_TYPE_1.getCode().equals(HADOOP_HOST.RUN_TYPE)) {
			
			// 跨平台提交
			conf.set("mapreduce.app-submission.cross-platform", "true");
			// 运行在yarn上
			conf.set("mapreduce.framework.name", "yarn");
			// resourcemanager地址
			conf.set("yarn.resourcemanager.hostname", "hadoop000");
			// HDFS
			conf.set("fs.defaultFS", "hdfs://hadoop000:9000/");
			// 设置环境变量
			conf.set("yarn.app.mapreduce.am.env", "HADOOP_MAPRED_HOME=$HADOOP_HOME");
			conf.set("mapreduce.map.env", "HADOOP_MAPRED_HOME=$HADOOP_HOME");
			conf.set("mapreduce.reduce.env", "HADOOP_MAPRED_HOME=$HADOOP_HOME");
		}
        return conf;
	}
}
